package hldh.push

import java.util.UUID

import akka.actor._
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import akka.cluster.sharding.{ClusterSharding, ClusterShardingSettings, ShardRegion}
import akka.pattern.ask
import akka.persistence.PersistentActor
import akka.util.Timeout
import com.typesafe.config.ConfigFactory
import io.undertow.Undertow
import io.undertow.server.handlers.resource.ClassPathResourceManager
import io.undertow.server.{HttpHandler, HttpServerExchange}
import io.undertow.util.Headers
import io.undertow.websockets.WebSocketConnectionCallback
import io.undertow.websockets.core._
import io.undertow.websockets.spi.WebSocketHttpExchange
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._

import scala.concurrent.Await
import scala.concurrent.duration._


/**
 * Created by jf on 16/1/23.
 */

object PushServer extends App {

  import io.undertow.Handlers._

  val port = if (args != null && args.length > 0) args.apply(0) else "2552"

  val wsPort = ("1" + port).toInt

  val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" + port).
    withFallback(ConfigFactory.load())

  implicit val actorSystem = ActorSystem("push-server", config)
  implicit val timeout = Timeout(15.seconds)

  val extractEntityId: ShardRegion.ExtractEntityId = {
    case c: NewConnect => (c.id, c)
    case p: Push => (p.id, p)
    case w: WhereIAm => (w.id, w)
  }

  val extractShardId: ShardRegion.ExtractShardId = {
    case c: NewConnect => (math.abs(c.id.hashCode) % 100).toString
    case p: Push => (math.abs(p.id.hashCode) % 100).toString
    case w: WhereIAm => (math.abs(w.id.hashCode) % 100).toString
  }

  //  startupSharedJournal(actorSystem, startStore = port == "2551", path =
  //    ActorPath.fromString("akka.tcp://push-server@127.0.0.1:2551/user/store"))

  //  val pushServer = actorSystem.actorOf(Props.apply(classOf[PushServer]), "push")

  val pushServer = ClusterSharding(actorSystem).start(
    typeName = "push",
    entityProps = Props.apply(classOf[PushServer], "127.0.0.1:" + wsPort + "/ws"),
    settings = ClusterShardingSettings(actorSystem),
    extractEntityId = extractEntityId,
    extractShardId = extractShardId)

  Undertow.builder().addHttpListener(wsPort, "127.0.0.1")
    .setHandler(path().addPrefixPath("/ws", websocket(new WebSocketConnectionCallback() {
    override def onConnect(exchange: WebSocketHttpExchange, channel: WebSocketChannel): Unit = {
      val params = exchange.getRequestParameters.get("pushId")
      val pushId = if (params == null || params.isEmpty) UUID.randomUUID().toString.replaceAll("-", "") else params.get(0)
      pushServer ! NewConnect(pushId, exchange, channel)
    }
  }))
    .addPrefixPath("/auth", path(new HttpHandler {
    override def handleRequest(exchange: HttpServerExchange): Unit = {
      exchange.getResponseHeaders.add(Headers.CONTENT_TYPE, "application/json;")

      // should auth with appId and appSecret and generate the pushId prefix with appId.
      val pushId = UUID.randomUUID().toString.replaceAll("-", "")

      var json = ("pushId" -> pushId) ~ ("timestamp" -> System.currentTimeMillis())
      val response = exchange.getResponseSender
      val where = pushServer ? WhereIAm(pushId)
      val address = Await.result(where, Duration.Inf)
      address match {
        case Here(address) => json = json ~ ("address" -> address)
      }
      response.send(compact(render(json)))
    }
  }))
    .addPrefixPath("/push", path(new HttpHandler {
    override def handleRequest(exchange: HttpServerExchange): Unit = {
      val params = exchange.getQueryParameters
      val msg = params.get("msg")
      val to = params.get("to")
      pushServer ! Push(to.getLast, msg.getLast)
      exchange.getResponseSender.close()
    }
  }))
    .addPrefixPath("/", resource(new ClassPathResourceManager(PushServer.getClass.getClassLoader, "webapp")).addWelcomeFiles("index.html"))
    ).build().start()


  //  def startupSharedJournal(system: ActorSystem, startStore: Boolean, path: ActorPath): Unit = {
  //    // Start the shared journal one one node (don't crash this SPOF)
  //    // This will not be needed with a distributed journal
  //    if (startStore)
  //      system.actorOf(Props[SharedLeveldbStore], "store")
  //    // register the shared journal
  //    import system.dispatcher
  //    implicit val timeout = Timeout(15.seconds)
  //    val f = system.actorSelection(path) ? Identify(None)
  //    f.onSuccess {
  //      case ActorIdentity(_, Some(ref)) => SharedLeveldbJournal.setStore(ref, system)
  //      case _ =>
  //        system.log.error("Shared journal not started at {}", path)
  //        system.terminate()
  //    }
  //    f.onFailure {
  //      case _ =>
  //        system.log.error("Lookup of shared journal at {} timed out", path)
  //        system.terminate()
  //    }
  //  }

}


class PushServer(wxPath: String) extends PersistentActor with ActorLogging {

  val cluster = Cluster(context.system)

  val nodesHash = new ConsistencyHash[Address]()

  override def preStart() = cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberEvent], classOf[UnreachableMember])

  override def postStop() = cluster.unsubscribe(self)

  private case class State(id: String, pushed: Boolean) {
    def update(evt: Event): State = evt match {
      case Push(id, msg) => copy(id = id, pushed = false)
      case Pushed(id) => copy(id = id, pushed = true)
    }
  }

  private var state = State("", pushed = false)

  override def receiveRecover: Receive = {
    case e: Event => state = state.update(e)
  }

  override def receiveCommand: Receive = {
    case x: WhereIAm =>
      println(x)
      sender() ! Here(wxPath)
    case x: NewConnect =>
      val webSocketServer = context.actorOf(Props.apply(classOf[WebSocketServer], x.channel), x.id)
      x.channel.getReceiveSetter.set(new AbstractReceiveListener() {
        override def onFullTextMessage(channel: WebSocketChannel, message: BufferedTextMessage) = {
          webSocketServer ! message
        }

        override def onCloseMessage(cm: CloseMessage, channel: WebSocketChannel): Unit = {
          super.onCloseMessage(cm, channel)
          webSocketServer ! cm
        }
      })
      x.channel.resumeReceives()

    case x: Push =>
      log.info("Push : {}", x)
      persist(Pushed(x.id)) { evt =>
        state = state.update(evt)
        context.child(x.id).getOrElse(sender()) ! PushMessage(x.message)
      }



    case MemberUp(member) =>
      log.info("node [{}] joined", member.address)
      nodesHash.add(member.address)
    case MemberRemoved(member, previousStatus) => nodesHash.remove(member.address)
  }

  override def persistenceId: String = "push-server-" + self.path.name

}

case class NewConnect(id: String, exchange: WebSocketHttpExchange, channel: WebSocketChannel)

case class WhereIAm(id: String)

case class Here(address: String)
